package com.sunzm.flink.datastream.java.operator;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JavaConnectDemo {
    private static final Logger logger = LoggerFactory.getLogger(JavaConnectDemo.class);
    private static boolean isLocal = true;

    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        if (isLocal) {
            env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        }

        DataStreamSource<String> dataStream1 = env.socketTextStream("82.156.210.70", 9999);
        DataStreamSource<String> dataStream2 = env.socketTextStream("82.156.210.70", 8888);


        ConnectedStreams<String, String> connectedStreams = dataStream1.connect(dataStream2);

        DataStream<String> mapedDataStream = connectedStreams.map(new CoMapFunction<String, String, String>() {

            @Override
            public String map1(String value) throws Exception {
                // 处理流1的数据
                logger.info("map1： " + value);

                return value;
            }

            @Override
            public String map2(String value) throws Exception {
                // 处理流1的数据
                logger.info("map2： " + value);

                return value;
            }
        });

        mapedDataStream.print("Map DataStream");

        DataStream<String> flatMapedDS = connectedStreams.flatMap(new CoFlatMapFunction<String, String, String>() {

            @Override
            public void flatMap1(String value, Collector<String> out) throws Exception {
                if (StringUtils.isNotBlank(value)) {
                    logger.info("flatMap1 : " + value);
                    //数据不为空才发出，map无法实现这个功能
                    out.collect(value);
                } else {
                    logger.info("flatMap1 收到空数据");
                }
            }

            @Override
            public void flatMap2(String value, Collector<String> out) throws Exception {
                if (StringUtils.isNotBlank(value)) {
                    logger.info("flatMap2 : " + value);
                    //数据不为空才发出，map无法实现这个功能
                    out.collect(value);
                } else {
                    logger.info("flatMap2 收到空数据");
                }
            }
        });

        flatMapedDS.print("FlatMap DataStream");

        //启动程序
        String className = Thread.currentThread().getStackTrace()[1].getClassName();
        String simpleName = className.substring(className.lastIndexOf(".") + 1);
        env.execute(simpleName);
    }
}
